package com.xzy.rabbit.demo.deadqueue.c;

import com.rabbitmq.client.*;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 死信队列_拒绝接收
 *
 * @author xzy
 * @date 2021/9/2811:20
 */
@Slf4j
@AllArgsConstructor
class ConsumerNormal implements Runnable {

    /**
     * 非持久化队列（RabbitMQ重启后队列消失）
     */
    private static final boolean DURABLE = false;
    /**
     * 独占队列（当前Connection独占）
     */
    private static final boolean EXCLUSIVE = false;
    /**
     * 自动删除队列（不再使用时自动删除）
     */
    private static final boolean AUTO_DELETE = false;

    private final Connection connection;
    private final String workExchangeName;
    private final String workRoutingKey;
    private final String workQueueName;
    private final long workTime;

    private final String deadExchangeName;
    private final String deadRoutingKey;

    public void consume() throws IOException {

        // 1.获取信道
        Channel channel = connection.createChannel();

        // 2.设置预取值(!!!)
        channel.basicQos(1);

        // 3.声明交换机（死信）
        channel.exchangeDeclare(deadExchangeName, BuiltinExchangeType.DIRECT);

        // 4.声明队列
        // 设置 死信交换机、死信路由键
        Map<String, Object> arguments = new HashMap<>(2);
        arguments.put("x-dead-letter-exchange", deadExchangeName);
        arguments.put("x-dead-letter-routing-key", deadRoutingKey);
        channel.queueDeclare(workQueueName, DURABLE, EXCLUSIVE, AUTO_DELETE, arguments);

        // 5.绑定交换机
        channel.queueBind(workQueueName, workExchangeName, workRoutingKey);

        // 6.接收消息
        String currentThread = Thread.currentThread().getName();
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            // 1.处理消息
            try {
                Thread.sleep(workTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 2.手动应答（随机）
            if (Math.random() < 0.5) {
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                log.info(currentThread + " 确认接收：`{}`<----{}\n", new String(message.getBody()), workRoutingKey);
            } else {
                // 拒绝接收该条消息，且不重新返回原队列
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
                log.info(currentThread + " 拒绝接收：`{}`<----{}\n", new String(message.getBody()), workRoutingKey);
            }

        };
        CancelCallback cancelCallback = consumerTag -> {
            log.info(currentThread + " 取消接收\nconsumeTag ==> {}\n", consumerTag);
        };
        channel.basicConsume(workQueueName, false, deliverCallback, cancelCallback);

        /*
         * 注意：关闭自动应答
         */
    }

    @Override
    public void run() {
        try {
            consume();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
